package com.heima.kafkastream;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.KeyValueStore;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 *上面 WordCountApplication类是官方源码 可读性差 so WordCountApplication393是重写的可读性好的
 */

public class WordCountApplication393 {

    public static void main(final String[] args) throws Exception {
        //1.构建属性对象
        Properties props = new Properties();
        //1.1 设置应用ID
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application");
        //1.2 设置broker服务器地址
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.137.136:9092");
        //1.3 设置Key的序列化器
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());//key序列化形式是string
        //1.4 设置val的序列化器
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());//value序列化形式是string
        //1.5 设置提交结果时间频率  1秒后提交 可跟时间统计窗口时间一致       kafkafrop界面不会立马收到过一会才能看到，因为它不会来一条提交一次而是kafkastream 积累一定数值过一段时间后统一做提交 so设置提交结果时间频率
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "10000");


        //2. 创建流构建对象
        StreamsBuilder builder = new StreamsBuilder();
        //2.1 设置源头话题"名称-----源处理器   --TODO 需要手动先创建该原始话题  数据形式 key:null val:你好 666 xxx
        KStream<String, String> textLines = builder.stream("TextLinesTopic393-time");//监听原始话题数据   数据源源不断进入此话题so监听是实时的   KStream<key的类型, 值的类型>
     //记得kafkadrop界面刷新 把新创建的话题要显示出来不然可能报错 记得把运行的原生代码WordCountApplication停掉因为和此页代码设置的应用id一致但应用id不能重复会报错
        //2.2 处理流数据-处理消息（数据）的val值 通过空格进行分割  流处理器              只处理value so用flatMapValues  且点开它看源码发现 是值映射到任意类型的新值 得返回新的结果 要ValueMapper<>类型so   new一个
        KStream<String, String> mapValues = textLines.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            /**                                                                            //返回迭代器得到处理值的所有数据  类型全改string类型
             * 将要处理数据流中值val
             * @param value
             * @return
             */
            @Override
            public Iterable<String> apply(String value) {
                System.out.println("采用空格切割处理前数据：" + value); //运行某个用户连续发不同的评论
                String[] s = value.split(" ");
                return Arrays.asList(s); //所有值封装到了Iterable<String>迭代器里 将数组转成了集合  传入string数组 再转成list集合
            }
        });
        //2.3 对上面处理器处理后数据 进行分组 值相同的放一组                 上面给的  数据形式 key:null val:你好 666 xxx   只能按value分组   要什么类型new什么类型
        KGroupedStream<String, String> groupedStream = mapValues.groupBy(new KeyValueMapper<String, String, String>() {
            /**                                                                            key的类型 value的类型 第三个位置是分组字段类型
             * 设置分组字段
             * @param key
             * @param value
             * @return
             */
            @Override
            public String apply(String key, String value) {
                System.out.println("进行分组，分组字段值：" + value);
                return value; //根据value分组so retrn value   返回分组字段类型 string 就是上面第三个string位置
            }
        });

        //2.3.1 对上面处理器处理后数据 设置时间窗口 统计最近10秒内数据   不是从开播开始累加 而是在某个时间段内的统计，过了这个时间就重新开始统计
        TimeWindowedKStream<String, String> timeWindowedKStream = groupedStream.windowedBy(TimeWindows.of(Duration.ofSeconds(10)));

                       //                                            Windowed<评论内容就是key>
        //2.4 对上面处理器处理后数据 进行统计分组后评论个数  经过分组加统计count后变 Key：Windowed<评论内容>  Val：数量
        KTable<Windowed<String>, Long> count = timeWindowedKStream.count();// 序列化时key 和value都是string 此处value变Long为避免出现乱码so 转成string


        //2.5 对上面处理器处理后数据 将分组统计数量Long类型改为String类型     把new 出来 ? 处类型改成string
        count.toStream().map(new KeyValueMapper<Windowed<String>, Long, KeyValue<String, String>>() {
            /** ....map(） 取到key 和value
             * 处理Val类型 重新设置Key 跟 Val 的类型
             * @param key 数据流中Key Windowed<评论内容>
             * @param value 数据流中Val
             * @return
             */
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, Long value) {
                System.out.println("进行转换值的类型，将Windowed<String>转为String");
                return new KeyValue<>(key.key(), value.toString());  //重新设置key 和value    long转string     key里是评论内容so从里面取内容key.key()
            }
        }).to("WordsWithCountsTopic393-time"); //结果话题  将结果实时产生到此 目标话题中

        //3.构建流式应用
        KafkaStreams streams = new KafkaStreams(builder.build(), props);

        //4.启动流式应用开始计算
        streams.start();
    }

}
